package server

import (
	"context"
	"git.huoys.com/middle-business/xds/internal/dao"
	"git.huoys.com/middle-business/xds/internal/model"
	"git.huoys.com/middle-end/kratos/pkg/conf/paladin"
	v2 "github.com/envoyproxy/go-control-plane/envoy/api/v2"
	core "github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
	endpoint "github.com/envoyproxy/go-control-plane/envoy/api/v2/endpoint"
	route "github.com/envoyproxy/go-control-plane/envoy/api/v2/route"
	discovery "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v2"
	"github.com/envoyproxy/go-control-plane/pkg/cache"
	"github.com/golang/protobuf/proto"
	"github.com/golang/protobuf/ptypes/any"
	"github.com/golang/protobuf/ptypes/duration"
	"log"
)

var (
	redisDao *dao.RedisDao
	k8s      *dao.K8S

	RouteConfigs *model.RouteConfig
	RouteConfigurations	*v2.RouteConfiguration

	ClusterConfigs *model.ClusterConfig
	Cluster []*v2.Cluster
)

func InitServerDeps() {
	redisDao = dao.GetRedisDao()
	k8s, _ = dao.NewK8sClient()

	initRoute()
	initCluster()

}

// XDSStreamServer common data type for xDS stream
type XDSStreamServer interface {
	Send(*v2.DiscoveryResponse) error
	Recv() (*v2.DiscoveryRequest, error)
	Context() context.Context
}

// Server struct will impl CDS, LDS, RDS & ADS
type Server struct{}

func (s *Server) DeltaAggregatedResources(server discovery.AggregatedDiscoveryService_DeltaAggregatedResourcesServer) error {
	panic("implement me")
}

func (s *Server) DeltaEndpoints(server v2.EndpointDiscoveryService_DeltaEndpointsServer) error {
	panic("implement me")
}

func (s *Server) DeltaRoutes(server v2.RouteDiscoveryService_DeltaRoutesServer) error {
	panic("implement me")
}

func (s *Server) DeltaListeners(server v2.ListenerDiscoveryService_DeltaListenersServer) error {
	panic("implement me")
}

func (s *Server) DeltaClusters(server v2.ClusterDiscoveryService_DeltaClustersServer) error {
	panic("implement me")
}

// BiDiStreamFor common bi-directional stream impl for cds,lds,rds
func (s *Server) BiDiStreamFor(xdsType string, stream XDSStreamServer) error {
	panic("implement me")
}

func (s *Server) LoadEndpoints(ctx context.Context, resourceNames []string) []*any.Any {
	anys := make([]*any.Any, 0)
	for _, name := range resourceNames {
		es := dao.EndpointMap[name]
		endpoints := []*endpoint.LbEndpoint{}
		for _, v := range es {
			endpoints = append(endpoints, &endpoint.LbEndpoint{
				HostIdentifier: &endpoint.LbEndpoint_Endpoint{Endpoint: v},
			})
		}
		assignment := &v2.ClusterLoadAssignment{
			ClusterName: name,
			Endpoints:   []*endpoint.LocalityLbEndpoints{{LbEndpoints: endpoints}},
		}
		data, err := proto.Marshal(assignment)
		if err != nil {
			log.Println("Marshal ClusterLoadAssignment fail, msg: ", err.Error())
			continue
		}
		a := &any.Any{
			TypeUrl: cache.EndpointType,
			Value:   data,
		}
		anys = append(anys, a)
	}
	return anys
}

func (s *Server) LoadRoutes() []*any.Any {
	anys := make([]*any.Any, 0)
	data, err := proto.Marshal(RouteConfigurations)
	if err != nil {
		log.Println("Marshal RouteConfiguration fail msg ", err.Error())
	}
	a := &any.Any{
		TypeUrl: cache.RouteType,
		Value:   data,
	}
	anys = append(anys, a)
	return anys
}

func (s *Server) LoadClusters() []*any.Any {
	anys := make([]*any.Any, 0)

	for _, c := range Cluster {
		marshal, err := proto.Marshal(c)
		if err != nil {
			log.Printf("Marshal cluster fail, msg %s\n", err.Error())
			continue
		}
		a := &any.Any{
			TypeUrl: cache.ClusterType,
			Value:   marshal,
		}
		anys = append(anys, a)
	}

	return anys
}

func IsValidSubscriber(req *v2.DiscoveryRequest) bool {
	return (len(req.Node.Cluster) > 0) && (len(req.Node.Id) > 0)
}

func initRoute() {
	cfg := &model.RouteConfig{}
	err := paladin.Watch("route.txt", cfg)
	if err != nil {
		log.Println(err.Error())
		return
	}
	routes := []*route.Route{}
	for _, r := range cfg.Routes{
		rou := &route.Route{Match: &route.RouteMatch{PathSpecifier: &route.RouteMatch_Prefix{Prefix: r.Match}},Action: &route.Route_Route{Route: &route.RouteAction{ClusterSpecifier: &route.RouteAction_Cluster{Cluster: r.Cluster}}}}
		routes = append(routes, rou)
	}
	host := &route.VirtualHost{
		Name:    "local_service",
		Domains: []string{"*"},
		Routes:  routes,
	}
	RouteConfigurations = &v2.RouteConfiguration{
		Name:         "local_route",
		VirtualHosts: []*route.VirtualHost{host},
	}

}

func initCluster() {
	cfg := &model.ClusterConfig{}
	err := paladin.Watch("cluster.txt", cfg)
	if err != nil {
		log.Println(err.Error())
		return
	}
	Cluster = []*v2.Cluster{}
	for _, c := range cfg.Clusters{
		clu := &v2.Cluster{Name: c.Name, ConnectTimeout: &duration.Duration{Seconds: c.ConnectTimeout}, ClusterDiscoveryType: &v2.Cluster_Type{Type: v2.Cluster_STRICT_DNS}, LbPolicy: v2.Cluster_ROUND_ROBIN, DnsLookupFamily: v2.Cluster_V4_ONLY, Hosts: []*core.Address{{Address: &core.Address_SocketAddress{SocketAddress: &core.SocketAddress{Address: c.Address, PortSpecifier: &core.SocketAddress_PortValue{PortValue: c.Port}}}}}}
		Cluster = append(Cluster, clu)
	}
}